package com.nervenets.general.redis;

import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.lang.reflect.ParameterizedType;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 基于redis的生产消息、消费消息
 * Created by joe on 2024/2/20.
 */
@Slf4j
public abstract class RedisSubPubManager<M> implements InitializingBean, MessageListener {
    private final Object queueLock = new Object();
    private final ExecutorService executorService = Executors.newFixedThreadPool(poolSize());
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    private Thread thread;

    @Override
    public void afterPropertiesSet() {
        if (null == redisTemplate) throw new IllegalStateException("this redis config is invalid");
        if (consumer()) {
            if (!persisted()) {
                redisTemplate.execute((RedisCallback<Void>) rc -> {
                    rc.subscribe(this, getChannel().getBytes(StandardCharsets.UTF_8));
                    return null;
                });
            } else {
                this.startPersistedTask();
            }
        }
        log.info("subscribe message `{}` success", getChannel());
    }

    private void startPersistedTask() {
        thread = ThreadUtil.newThread(() -> {
            synchronized (queueLock) {
                while (true) {
                    Long length = redisTemplate.execute((RedisCallback<Long>) rc -> rc.lLen(getChannel().getBytes(StandardCharsets.UTF_8)));
                    if (null == length || length == 0L) {
                        try {
                            //log.info("subscribe message `{}` waiting {}ms...", getChannel(), waitingTime());
                            queueLock.wait(waitingTime());
                        } catch (InterruptedException ignored) {
                        }
                    } else {
                        log.info("subscribe message `{}` length {}", getChannel(), length);
                        final ThreadPoolExecutor executor = (ThreadPoolExecutor) this.executorService;
                        //System.out.printf("active:%s completedTaskCount:%s taskCount:%s%n", executor.getActiveCount(), executor.getCompletedTaskCount(), executor.getTaskCount());
                        if (executor.getActiveCount() == 0) {
                            int threadSize = length > poolSize() ? poolSize() : length.intValue();
                            for (int i = 0; i < threadSize; i++) {
                                this.executorService.execute(() -> {
                                    final byte[] result = redisTemplate.execute((RedisCallback<byte[]>) rc -> rc.rPop(getChannel().getBytes(StandardCharsets.UTF_8)));
                                    if (null != result) {
                                        try {
                                            this.message(JSON.parseObject(result, ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0]));
                                        } catch (Exception e) {
                                            log.info("consume message failed: {} reason:{}", new String(result), e.getMessage());
                                        }
                                    }
                                });
                            }
                        } else {
                            log.info("subscribe message `{}` waiting last executor 1 SECOND", getChannel());
                            try {
                                TimeUnit.SECONDS.sleep(1);
                            } catch (InterruptedException e) {
                                //
                            }
                        }
                    }
                }
            }
        }, "RedisSubPubTask");
        thread.start();
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        this.message(JSON.parseObject(message.getBody(), ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0]));
    }

    public Long pub(M body) {
        if (null == redisTemplate) throw new IllegalStateException("this redis config is invalid");
        if (persisted()) {
            Long result = redisTemplate.execute((RedisCallback<Long>) rc -> rc.lPush(getChannel().getBytes(StandardCharsets.UTF_8), JSON.toJSONBytes(body)));
            ThreadUtil.execute(() -> {
                synchronized (queueLock) {
                    queueLock.notifyAll();
                }
            });
            return result;
        }
        return redisTemplate.execute((RedisCallback<Long>) rc -> rc.publish(getChannel().getBytes(StandardCharsets.UTF_8), JSON.toJSONBytes(body)));
    }

    /**
     * 持久化消息队列将以rPush和lPop来实现消息的消费
     * 非持久化的消息将以订阅的方式实现，当订阅者消失后，消息会丢失
     *
     * @return 消息是否持久化
     */
    public boolean persisted() {
        return true;
    }

    /**
     * @return 是否自己消费
     */
    public boolean consumer() {
        return true;
    }

    /**
     * 当持续没有取到消息消费时，最大的等待时长
     *
     * @return 最大等待时间
     */
    public int waitingTime() {
        return 60000;
    }

    public int poolSize() {
        return 10;
    }

    @PreDestroy
    public void destroy() {
        log.info("subscribe message `{}` destroy", getChannel());
        executorService.shutdown();
        if (null != thread) {
            thread.interrupt();
        }
    }

    protected abstract String getChannel();

    protected abstract void message(M body);
}
